#include <unistd.h>
#include <poll.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <limits.h>
#include <stdio.h>
#include "eloop.h"

#define errorinfo(name)  printf("%s failed: errno=%d error=%s\n", #name, errno, strerror(errno))


typedef struct _esource    esource_t;
typedef struct _io_source  io_source_t;
typedef struct _tmo_source tmo_source_t;
typedef enum _source_type  stype_t;
typedef struct _run_data   run_data_t;

struct _eloop {
  int         wakeup_fd;
  int         wait_fd;
  int         polling;
  pthread_mutex_t lock;
  esource_t  *sources;
};

enum _source_type {
  S_TYPE_TMO,
  S_TYPE_IO
};

struct _esource {
  int         type;
  esource_t  *next;
};

struct _io_source {
  int           type;
  esource_t    *next;

  int           fd;
  short         events;
  io_handler_t  handler;
  void         *data;
};

struct _tmo_source {
  int        type;
  esource_t *next;

  long long  uptime;
  int        expire;
  int        checked;
  int (*expire_cb)(eloop_t*, void*);
  void      *data;
};

struct _run_data {
  void (*cb)(eloop_t*, void *data);
  void *data;
};


static io_source_t* eloop_find_iosource_by_fd_unlock(eloop_t *loop, int fd);
static int eloop_remove_source_unlock(eloop_t *loop, esource_t *source);
static loop_event_t pollevent2loopevent(short pollevents);
static short loopevent2pollevent(loop_event_t loopevents);
static int timeout_once_cb(eloop_t *eloop, void *data);
static void eloop_source_append(eloop_t *eloop, esource_t *source);

static void
eloop_lock(eloop_t *eloop)
{
  if (pthread_mutex_lock(&eloop->lock))
  {
    errorinfo(pthread_mutex_lock);
    assert(0);
  }
}

static void
eloop_unlock(eloop_t *eloop)
{
  if (pthread_mutex_unlock(&eloop->lock))
  {
    errorinfo(pthread_mutex_unlock);
    assert(0);
  }
}

static long long
get_uptime(void)
{
  struct timespec ts;
  
  if (clock_gettime(CLOCK_MONOTONIC, &ts) == 0)
    return ts.tv_sec *1000 + ts.tv_nsec / 1000000;
  else
    errorinfo(clock_gettime);
  return 0;
}

eloop_t*
eloop_init(void)
{
  eloop_t *eloop;
  int  pipe_fd[2];

  eloop = calloc(1, sizeof(*eloop));
  if (pipe(pipe_fd))
  {
    errorinfo(pipe);
    free(eloop);
    return NULL;
  }
  eloop->wait_fd = pipe_fd[0];
  eloop->wakeup_fd = pipe_fd[1];
  eloop->polling = 0;

  pthread_mutex_init(&eloop->lock, NULL);
  return eloop;
}

void
eloop_run (eloop_t *eloop)
{
  struct pollfd *fds;
  int  nfds_cap = 16;
  int  nfds;
  int  timeout;
  int  pollret;
  int  i;
  esource_t *source;
  io_source_t *iosource;
  tmo_source_t *tmosource;
  
  fds = calloc(nfds_cap, sizeof(struct pollfd));
  while (1)
  {
    eloop_lock(eloop);
    nfds = 1;
    timeout = INT_MAX;
    for (source = eloop->sources; source; source = source->next)
    {
      if (source->type == S_TYPE_IO)
      {
        nfds += 1;
        if (nfds >= nfds_cap)
        {
          nfds_cap += 16;
          fds = realloc(fds, nfds_cap);
        }
        fds[nfds - 1].fd = ((io_source_t *)source)->fd;
        fds[nfds - 1].events = ((io_source_t *)source)->events;
        fds[nfds - 1].revents = 0;
      }
      else
      {
        tmosource = (tmo_source_t *)source;
        tmosource->checked = 0;
        if (tmosource->expire < timeout)
          timeout = tmosource->expire;
      }
    }
    fds[0].fd = eloop->wait_fd;
    fds[0].events = POLLIN;
    fds[0].revents = 0;

    if (timeout < 0)
      timeout = 0;
    if (timeout > 0)
      eloop->polling = 1;

    eloop_unlock(eloop);

    if (timeout > 0)
      pollret = poll(fds, nfds, timeout);
    else
      pollret = 0;
    if (pollret < 0)
    {
      errorinfo(poll);
      usleep(1500*1000);
      continue;
    }

    eloop_lock(eloop);
    eloop->polling = 0;
    for (i=1; i<nfds; i++)
    {
      if (fds[i].revents)
      {
        iosource = eloop_find_iosource_by_fd_unlock(eloop, fds[i].fd);
        if (iosource)
        {
          eloop_unlock(eloop);
          iosource->handler(eloop, iosource->fd,
                            pollevent2loopevent(fds[i].revents),
                            iosource->data);
          eloop_lock(eloop);
        }
      }
    }
    if (fds[0].revents)
    {
      char c;
      read(fds[0].fd, &c, 1);
    }
tmo_recheck:
    source = eloop->sources;
    while (source)
    {
      if (source->type == S_TYPE_TMO)
      {
        int  remove = 0;
        long long curtime;
        long  delta;
        tmosource = (tmo_source_t *)source;
        if (tmosource->checked)
          goto next_source;
        tmosource->checked = 1;
        curtime = get_uptime();
        delta = curtime - tmosource->uptime;
        if (delta < 0) delta += LONG_MAX;
        if (delta >= tmosource->expire)
        {
          tmosource->uptime = curtime;
          eloop_unlock(eloop);
          remove = !tmosource->expire_cb(eloop, tmosource->data);
          eloop_lock(eloop);
          if (remove)
            eloop_remove_source_unlock(eloop, source);
          goto tmo_recheck;
        }
      }
next_source:
      source = source->next;
    }
    eloop_unlock(eloop);
  }
}

int
eloop_add_timeout(eloop_t *eloop, int timeout, int (*timeoutcb)(eloop_t *eloop, void *data), void *data)
{
  tmo_source_t *tmosource = calloc(1, sizeof(*tmosource));
  if (!tmosource)
    return 0;

  tmosource->type = S_TYPE_TMO;
  tmosource->next = NULL;
  tmosource->uptime = get_uptime();
  tmosource->expire = timeout;
  tmosource->checked = 1;//initial checked
  tmosource->expire_cb = timeoutcb;
  tmosource->data = data;

  eloop_source_append(eloop, (esource_t*)tmosource);

  return (int)tmosource;
}

int
eloop_add_io_source(eloop_t *eloop, int fd, loop_event_t events, io_handler_t handler, void *data)
{
  io_source_t *iosource = calloc(1, sizeof(*iosource));
  if (!iosource)
    return 0;

  iosource->type = S_TYPE_IO;
  iosource->next = NULL;
  iosource->fd = fd;
  iosource->events = loopevent2pollevent(events);
  iosource->handler = handler;
  iosource->data = data;

  eloop_source_append(eloop, (esource_t*)iosource);

  return (int)iosource;
}

void
eloop_add_run(eloop_t *eloop, void (*runcb)(eloop_t *eloop, void *data), void *data)
{
  run_data_t *run = calloc(1, sizeof(*run));
  run->cb = runcb;
  run->data = data;
  eloop_add_timeout(eloop, 0, timeout_once_cb, run);
}

int
eloop_remove_source(eloop_t *eloop, int id)
{
  esource_t *source = (esource_t *)id;
  int ret;

  eloop_lock(eloop);
  ret = eloop_remove_source_unlock(eloop, source);
  eloop_unlock(eloop);

  return ret;
}

int
eloop_remove_timeout_by_cb(eloop_t *eloop, int (*timeoutcb)(eloop_t *eloop, void *data))
{
  esource_t **walk;
  int found = 0;

  eloop_lock(eloop);
  walk = &eloop->sources;
  while (*walk)
  {
    if ((*walk)->type == S_TYPE_TMO)
    {
      tmo_source_t *tmosource = (tmo_source_t *)*walk;
      if (tmosource->expire_cb == timeoutcb)
      {
        found = 1;
        *walk = (*walk)->next;
        free (tmosource);
        break;
      }
    }
    walk = &(*walk)->next;
  }
  eloop_unlock(eloop);

  return !found;
}

int
eloop_remove_io_source_by_fd(eloop_t *eloop, int fd)
{
  esource_t **walk;
  int found = 0;

  eloop_lock(eloop);
  walk = &eloop->sources;
  while (*walk)
  {
    if ((*walk)->type == S_TYPE_IO)
    {
      io_source_t *iosource = (io_source_t *)*walk;
      if (iosource->fd == fd)
      {
        found = 1;
        *walk = (*walk)->next;
        free (iosource);
        break;
      }
    }
    walk = &(*walk)->next;
  }
  eloop_unlock(eloop);

  return !found;

}

static int
timeout_once_cb(eloop_t *eloop, void *data)
{
  run_data_t *run = data;
  run->cb(eloop, run->data);
  free(run);

  return 0;
}

static int
eloop_remove_source_unlock(eloop_t *eloop, esource_t *source)
{
  esource_t **walk = &eloop->sources;

  while(*walk)
  {
    if (*walk == source)
    {
      *walk = (*walk)->next;
      free(source);
      return 0;
    }
    walk = &(*walk)->next;
  }

  return 1;
}

static io_source_t*
eloop_find_iosource_by_fd_unlock(eloop_t *eloop, int fd)
{
  esource_t *source;

  for (source=eloop->sources; source; source = source->next)
  {
    if (source->type == S_TYPE_IO)
    {
      io_source_t *iosource = (io_source_t *)source;
      if (iosource->fd == fd)
        return iosource;
    }
  }

  return NULL;
}

static void
eloop_source_append(eloop_t *eloop, esource_t *source)
{
  eloop_lock(eloop);
  if (!eloop->sources)
  {
    eloop->sources = source;
  }
  else
  {
    esource_t *s = eloop->sources;
	while(s->next) s = s->next;
	s->next = source;
  }
  if (eloop->polling)
  {
    char c = 'a';
    write(eloop->wakeup_fd, &c, 1);
    eloop->polling = 0;
  }
  eloop_unlock(eloop);
}

static loop_event_t
pollevent2loopevent(short pollevents)
{
  loop_event_t le = 0;
  if (pollevents & POLLIN) le |= EVENT_IN;
  if (pollevents & POLLOUT) le |= EVENT_OUT;
  if (pollevents & POLLERR) le |= EVENT_ERR;
  if (pollevents & POLLPRI) le |= EVENT_IN;
  if (pollevents & POLLNVAL) le |= EVENT_NVAL;

  return le;
}

static short
loopevent2pollevent(loop_event_t loopevents)
{
  short pe = 0;
  if (loopevents & EVENT_IN) pe |= POLLIN;
  if (loopevents & EVENT_OUT) pe |= POLLOUT;
  if (loopevents & EVENT_ERR) pe |= POLLERR;
  if (loopevents & EVENT_NVAL) pe |= POLLNVAL;

  return pe;
}

